package com.java.learn.wordcount;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

/**
 * @Copyright: Shanghai Definesys Company.All rights reserved.
 * @Description: java 执行wordcount
 * @author: chuhaitao
 * @since: 2019/2/17 21:08
 * @history: 1.2019/2/17 created by chuhaitao
 */
public class JavaWordCount {

    public static void main(String[] args) {
        //创建sparkConf
        SparkConf conf = new SparkConf();
        conf.setAppName("JavaWordCount");//.setMaster("loacal");
        //创建java sc
        JavaSparkContext jsc = new JavaSparkContext(conf);

        final JavaRDD<String> lines = jsc.textFile(args[0]);

        final JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String line) throws Exception {
                String[] words = line.split(" ");
                return Arrays.asList(words).iterator();
            }
        });
        //组装单词
        final JavaPairRDD<String, Integer> wordsAndOne = words.mapToPair(new PairFunction<String, String, Integer>() {
            //java没有元祖，只有定义的类
            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<String, Integer>(word, 1);
            }
        });

        //聚合同一个key

        final JavaPairRDD<String, Integer> reduced = wordsAndOne.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });

        //java中排序只能按照key，所以先反过来
        final JavaPairRDD<Integer, String> reduced2 = reduced.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
            @Override
            public Tuple2<Integer, String> call(Tuple2<String, Integer> tuple2) throws Exception {
                return tuple2.swap();
            }
        });
        //排序
        final JavaPairRDD<Integer, String> sorted = reduced2.sortByKey(false);
        final JavaPairRDD<String, Integer> sorted2 = sorted.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Tuple2<Integer, String> integerStringTuple2) throws Exception {
                return integerStringTuple2.swap();
            }
        });

        sorted2.saveAsTextFile(args[1]);
        jsc.close();
    }
}
